package com.example.testkafkaproto.producer.config;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;

@Configuration
public class ProducerConfiguration {

    private static final String BROKER_LIST = "192.168.11.202:9092";

    @Bean
    public KafkaProducer<String,String> kafkaProducer(){


        Properties props = new Properties();
        //"bootstrap.servers"
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BROKER_LIST);
        //"key.serializer"
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //"value.serializer"
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        //"client.id"不配默认会生成
        props.put(ProducerConfig.CLIENT_ID_CONFIG,"testProducer");
        //"retries"重试次数(内部可处理异常)
        props.put(ProducerConfig.RETRIES_CONFIG,3);
        //"enable.idempotence"幂等
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,true);
        //"max.in.flight.requests.per.connection"一个发送链接的缓存请求数
        props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION,1);
        //"acks"
        props.put(ProducerConfig.ACKS_CONFIG,"-1");
        //生产者拦截器
//        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG,"com.example.testkafkaproto.producer.interceptor.TestProducerInterceptor");
        //"compression.type"消息压缩类型
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,"gzip");
        KafkaProducer<String,String> kafkaProducer = new KafkaProducer(props);
        return kafkaProducer;
    }
}
